package com.li.flink.app

import com.alibaba.fastjson.JSON
import com.li.flink.domain.{Access, TopNAppRes}
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.api.java.functions.KeySelector
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

object TopNAppV1 {

  def main(args: Array[String]): Unit = {
    val inputPath = "/Users/lijiacen/Downloads/access.json"
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val textStream = env.readTextFile(inputPath)
    val mapStream: DataStream[Access] = textStream.map(m => {
      try {
        val obj: Access = JSON.parseObject(m, classOf[Access])
        obj
      } catch {
        case ex: Exception => print("json数据解析失败：" + m)
          null
      }
    }).filter(m => {
      m != null
    })
    val cleanStream = mapStream.assignTimestampsAndWatermarks(
      new BoundedOutOfOrdernessTimestampExtractor[Access](Time.seconds(20)) {
        override def extractTimestamp(t: Access) = t.getTime
      }).filter(m => {
      m.getEvent != "startup"
    })

    val keySelector = new KeySelector[Access, Tuple3[String, String, String]] {
      override def getKey(in: Access): Tuple3[String, String, String] = {
        (in.getEvent, in.getProduct.getCategory, in.getProduct.getName)
      }
    }

    val keySelector2 = new KeySelector[TopNAppRes, Tuple4[String, String, Long, Long]] {
      override def getKey(in: TopNAppRes): Tuple4[String, String, Long, Long] = {
        (in.getEvent, in.getCategory, in.getStart, in.getEnd)
      }
    }

    val windowStream = cleanStream.keyBy(keySelector).window(SlidingEventTimeWindows.of(Time.minutes(5), Time.minutes(1)))
    val aggStream = windowStream.aggregate(new topNAggregateFunction(), new topNWindowFunction())
    val keyedStream = aggStream.keyBy(keySelector2).process(new
        KeyedProcessFunction[Tuple4[String, String, Long, Long], TopNAppRes, List[TopNAppRes]] {

      private var listState: ListState[TopNAppRes] = _

      override def open(parameters: Configuration): Unit = {
        listState = getRuntimeContext().getListState(new ListStateDescriptor[TopNAppRes]("cnt-state", classOf[TopNAppRes]))
      }

      override def processElement(i: TopNAppRes, context: KeyedProcessFunction[(String, String, Long, Long), TopNAppRes, List[TopNAppRes]]#Context, collector: Collector[List[TopNAppRes]]): Unit = {
        listState.add(i)
        context.timerService().registerEventTimeTimer(i.getEnd + 1)
      }


      override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[(String, String, Long, Long), TopNAppRes, List[TopNAppRes]]#OnTimerContext, out: Collector[List[TopNAppRes]]): Unit = {
        val it = listState.get().iterator()
        val list: List[TopNAppRes] = List()
        var i = 0
        while (it.hasNext) {
          if (i < 3) {
            i = i + 1
            //            it.next
            println(it.next)
          } else {
            it.next
          }
        }
        //        println(list.size)
        out.collect(list)
      }
    })
    //    keyedStream.print()
    env.execute("TopNAppV1")
  }
}

class topNAggregateFunction extends AggregateFunction[Access, Long, Long] {
  override def createAccumulator(): Long = 0L

  override def add(in: Access, acc: Long): Long = acc + 1

  override def getResult(acc: Long): Long = acc

  override def merge(acc: Long, acc1: Long): Long = ???
}

class topNWindowFunction extends ProcessWindowFunction[Long, TopNAppRes, Tuple3[String, String, String], TimeWindow] {
  override def process(key: (String, String, String), context: Context, elements: Iterable[Long], out: Collector[TopNAppRes]): Unit = {
    val counts = elements.iterator.next()
    val start = context.window.getStart()
    val end = context.window.getEnd()
    out.collect(new TopNAppRes(key._1, key._2, key._3, counts, start, end))
  }
}